package com.kingsoft.dc.khaos.module.spark.util;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.LinkedList;
import java.util.List;
import java.util.UUID;

/**
 * Created by WANGYING15 on 2019/7/29.
 */
public class RocketMQProducer {
    private static Logger log = LoggerFactory.getLogger(RocketMQProducer.class);

    public static void main(String[] args) {

        List<String> traverseList = new LinkedList<String>(){{
            add("gzu ");
            add("pyu ");
            add("thinking ");
            add("in ");
            add("java ");
        }};

        syncSend("192.168.158.56:9876", "producer111", "CheckTopicTest", "TagA", traverseList);
    }

    public static void syncSend(String nameServer, String groupName, String topic, String tags, List<String> messages) {

        //屏蔽mq自身的日志配置
        //System.setProperty("rocketmq.client.log.loadconfig", "false");

        System.setProperty("rocketmq.client.logUseSlf4j", "true");

        //声明并初始化一个producer
        //需要一个producer group名字作为构造方法的参数，这里为producer111
        DefaultMQProducer producer = new DefaultMQProducer(groupName);

        //设置NameServer地址,此处应改为实际NameServer地址，多个地址之间用；分隔
        //NameServer的地址必须有，但是也可以通过环境变量的方式设置，不一定非得写死在代码里
        producer.setNamesrvAddr(nameServer);
        producer.setInstanceName(UUID.randomUUID().toString());

        //调用start()方法启动一个producer实例
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        //发送10条消息到Topic为TopicTest，tag为TagA，消息内容为“Hello RocketMQ”拼接上i的值
        for (int i = 0; i < messages.size(); i++) {
            try {
                Message msg = new Message(topic,// topic
                        tags,// tag
                        messages.get(i).getBytes(RemotingHelper.DEFAULT_CHARSET)// body
                );

                //调用producer的send()方法发送消息
                //这里调用的是同步的方式，所以会有返回结果
                SendResult sendResult = producer.send(msg);

                //打印返回结果，可以看到消息发送的状态以及一些相关信息
                //System.out.println(sendResult);
                log.info("发送的消息ID:" + sendResult.getMsgId() +"--- 发送消息的状态：" + sendResult.getSendStatus());

            } catch (Exception e) {
                e.printStackTrace();
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
            }
        }

        //发送完消息之后，调用shutdown()方法关闭producer
        producer.shutdown();
    }
}
